package com.yujiahao;

import org.apache.flink.kinesis.shaded.com.amazonaws.ClientConfiguration;
import org.apache.flink.kinesis.shaded.com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
import org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.AmazonKinesis;
import org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.AmazonKinesisClientBuilder;
import org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.model.*;
import org.junit.Before;
import org.junit.Test;

import java.math.BigInteger;
import java.util.ArrayList;
import java.util.List;

/**
 * @Description: 拆分合并分片（分区）
 * @Author: yujiahao
 * @Date: 2023/8/2 16:38
 * @Version: 1.0
 */

public class OperationShards {


    private static AmazonKinesis client;
    private String streamName = "test-kds-api";
    private String sequenceNumberOfPreviousRecord;
    DescribeStreamRequest describeStreamRequest;

    /**
     * 创建客户端对象
     */
    @Before
    public void createClient() {
        //构建客户端对象
        AmazonKinesisClientBuilder clientBuilder = AmazonKinesisClientBuilder.standard();
        ClientConfiguration clientConfiguration = new ClientConfiguration();
        clientBuilder.setCredentials(new DefaultAWSCredentialsProviderChain());
        clientBuilder.setClientConfiguration(clientConfiguration);
        client = clientBuilder.build();
        describeStreamRequest = new DescribeStreamRequest().withStreamName(streamName);
    }

    /**
     * todo 应该属于客户端操作，对流本身的一个设置，也不能说是生产者，先放着吧
     *  这个其实蛮有用的，如果当前分片来不及处理，可能会造成数据积压处理不过来，那么拆分是个有效的方法
     * 拆分分片，需要指定如何将父分片中的哈希键值重新分配到子分片。当然了你父分片或者流的状态得是可用的状态 active
     * ①首先调用 splitShards() 方法进行分片拆分，并等待一段时间以确保分片拆分完成，
     * ②然后通过调用 describeStream 方法来验证分片拆分的结果。
     */
    @Test
    public void splitShard() {

        //拆分操作
        try {
            // 执行分片拆分操作
            splitShards();

            // 等待一段时间，以确保分片拆分完成
            Thread.sleep(60000);

            // 验证分片拆分后的情况 打印最终的分片信息看看是不是对的
            DescribeStreamResult describeStreamResult = client.describeStream(describeStreamRequest);
            List<Shard> shards = describeStreamResult.getStreamDescription().getShards();
            System.out.println("切分后的分片数: " + shards.size());

            // 打印切分后的分片信息
            for (Shard shard : shards) {
                System.out.println("分片 ID: " + shard.getShardId());
                System.out.println("哈希键范围: [" + shard.getHashKeyRange().getStartingHashKey() + " - " + shard.getHashKeyRange().getEndingHashKey() + "]");
            }
        } catch (Exception e) {
            System.err.println("执行分片拆分时发生错误: " + e.getMessage());
        }

    }

    /**
     * splitShards() 方法：该方法用于分片拆分操作。splitShard()的辅助方法，主要在这里面
     * 解题思路：
     * ①客户端就不说了，得有，它首先调用 describeStream 方法来获取当前数据流中的所有分片信息。
     * ②然后，它遍历每个分片，并计算新的起始哈希键以进行分片拆分。
     * ③接着，它调用 splitShard 方法进行分片拆分，并使用 waitForSplitCompletion() 方法等待分片拆分的完成。
     */
    private void splitShards() {
        try {
            //todo 1、获取流中的的详细信息的到一个结果
            DescribeStreamResult describeStreamResult = client.describeStream(describeStreamRequest);
            //todo 2、然后从DescribeStreamResult中得到分片的信息得到一个list集合
            List<Shard> shards = describeStreamResult.getStreamDescription().getShards();
            System.out.println("数据流中的分片数: " + shards.size());
            //todo 3、遍历每个分区的哈希值得到一个分区的范围
            for (Shard shard : shards) {
                String startingHashKey = shard.getHashKeyRange().getStartingHashKey();
                String endingHashKey = shard.getHashKeyRange().getEndingHashKey();

                System.out.println("分片 ID: " + shard.getShardId());
                System.out.println("哈希键范围: [" + startingHashKey + " - " + endingHashKey + "]");

                // todo 3.1 计算新的起始哈希键用于分裂分片 这里是对半分，实际也差不多对半分多一点，我在这里提供一个1：2分的例子

                /**
                 * 这段代码用于计算新的起始哈希键 newStartingHashKey，以进行分片拆分操作。
                 * 0️⃣ startingHashKey 和 endingHashKey 是分片的起始哈希键和结束哈希键，它们是字符串类型。
                 * ① 首先，将这两个字符串类型的哈希键转换为 BigInteger 类型的对象。BigInteger 是 Java 中用于表示大整数的类，可以进行高精度计算。
                 * ② 接着，使用 add 方法对 startingHashKeyBigInt 和 endingHashKeyBigInt 进行加法运算，得到它们的和，是在后面计算中点用。
                 * ③ 然后，使用 divide 方法将上一步得到的中点除以 2，得到新的起始哈希键。这样做的目的是找到原有分片的中点，从而得到一个新的起始哈希键，用于将原有分片进行拆分。
                 * ④ 最后，将新的起始哈希键转换为字符串类型，并赋值给 newStartingHashKey 变量，用于后续的分片拆分操作。
                 * ⑤ 通过这个过程，我们能够得到一个新的起始哈希键 newStartingHashKey，用于将原有分片进行拆分，从而实现数据流的分片扩展和数据处理能力的提升。
                 todo 如果将原有的分片拆分成 1：2 则使用下面的，理解理解
                 // 计算旧分片的哈希键范围大小
                 BigInteger rangeSize = endingHashKeyBigInt.subtract(startingHashKeyBigInt);

                 // 计算新分片的哈希键范围大小（旧分片范围的 1/2）
                 BigInteger newRangeSize = rangeSize.divide(new BigInteger("2"));

                 // 计算新的起始哈希键
                 BigInteger newStartingHashKeyBigInt = startingHashKeyBigInt.add(newRangeSize);

                 // 将新的起始哈希键转换为字符串表示
                 String newStartingHashKey = newStartingHashKeyBigInt.toString();

                 */
                BigInteger startingHashKeyBigInt = new BigInteger(startingHashKey);
                BigInteger endingHashKeyBigInt = new BigInteger(endingHashKey);
                String newStartingHashKey = startingHashKeyBigInt.add(endingHashKeyBigInt).divide(new BigInteger("2")).toString();

                // todo 3.2 执行分片分裂操作
                SplitShardRequest splitShardRequest = new SplitShardRequest()
                        .withStreamName(streamName)
                        .withShardToSplit(shard.getShardId())
                        .withNewStartingHashKey(newStartingHashKey);
                // todo 3.3 这里就是切分完成的流的信息了
                SplitShardResult splitShardResult = client.splitShard(splitShardRequest);
                System.out.println("正在拆分分片: " + shard.getShardId());
                System.out.println("新的起始哈希键: " + newStartingHashKey);

                // todo 3.4 等待分片分裂完成 这里调用了一个方法这个方法就是等待分片完成，如果单纯的想拆分上面几行就足以
                waitForSplitCompletion(streamName, shard.getShardId());
            }
        } catch (Exception e) {
            System.err.println("分片分裂时发生错误: " + e.getMessage());
        }
    }

    /** todo 拆分分片的辅助方法二
     * waitForSplitCompletion() 方法：该方法用于等待分片拆分的完成。
     * ①它使用 while 循环来轮询 describeStream 方法，检查目标分片的状态，直到分片拆分完成为止。
     * ②在目标分片的 SequenceNumberRange 中，endingSequenceNumber 不为空时表示分片拆分完成。
     * @param streamName 流的名称 这个在这里可以不用，因为里面流描述对象已经在初始化 @Before声明了，不想优化了就这样吧这个看情况，如果没有声明就得要这个注意下
     * @param shardId 分片(分区)id（在这里是已经切割后的流的分区id）
     */
    private void waitForSplitCompletion(String streamName, String shardId) {
        try {
            System.out.println("等待分片分裂完成...");

            while (true) {
                // todo 1、轮询 DescribeStream 操作检查分片分裂是否完成
                DescribeStreamResult describeStreamResult = client.describeStream(describeStreamRequest);
                List<Shard> shards = describeStreamResult.getStreamDescription().getShards();

                //todo 2、 查找目标分片,获取当前从流里面查到的分片ID和上面切割完成的传进来的分片id对比，如果匹配说明已经切分完成生效了
                Shard targetShard = null;
                for (Shard shard : shards) {
                    if (shard.getShardId().equals(shardId)) {
                        targetShard = shard;
                        break;
                    }
                }

                //todo 3、 检查目标分片状态
                if (targetShard == null || targetShard.getSequenceNumberRange().getEndingSequenceNumber() != null) {
                    break; // 分片拆分完成，跳出循环
                }

                // 等待一段时间后再次检查
                Thread.sleep(5000);
            }

            System.out.println("分片分裂完成.");
        } catch (Exception e) {
            System.err.println("等待分片分裂完成时发生错误: " + e.getMessage());
        }
    }


    /**
     * 分片的合并，这里每次合并一个
     * todo  注意：
     * 要合并两个分片，分片必须相邻。如果两个分片的哈希键范围的并集形成一个没有间隙的连续集，则两个分片被视为相邻。例如，假设您有两个分片，一个分片的哈希键范围为 276...381，另一个分片的哈希键范围为 382...454。
     * 然后才可以将这两个分片合并为一个分片，该分片的哈希键范围为 276...454。
     * 再举一个例子，假设两个分片，一个分片的哈希键范围为 276..381，另一个分片的哈希键范围为 455...560。那可无法合并这两个分片，因为这两个分片之间会有一个或多个分片覆盖范围 382..454。
     * 流中所有分片的集合OPEN（作为一个组）始终跨越 MD5 哈希键值的整个范围。
     *
     * todo 几个点：
     *  ① 因此要识别要合并的候选分片，应该过滤掉处于某种CLOSED状态的所有分片。
     *  ② 分片的结束序列号OPEN为。可以使用以下方法测试分片的结束序列号：CLOSED null
     *      shard.getSequenceNumberRange().getEndingSequenceNumber() == null
     *  ③ 按照每个分片支持的最高哈希键值对剩余分片进行排序。可以使用以下方法检索该值：
     *      shard.getHashKeyRange().getEndingHashKey();
     *  ④ describeStream 获取的 shards 列表包含了所有状态的分区，包括 ACTIVE、UPDATING 和 CLOSED 等状态的分区
     *  ⑤ 合并多个分片可能会涉及到流的数据迁移，因此请谨慎操作，确保在执行合并操作之前对数据流和应用程序进行充分的测试和验证。
     */
    @Test
    public void mergeShard() {
        // 获取数据流的描述信息
        DescribeStreamResult describeStreamResult = client.describeStream(describeStreamRequest);
        StreamDescription streamDescription = describeStreamResult.getStreamDescription();

        // 检查流状态是否为 ACTIVE
        if (!"ACTIVE".equals(streamDescription.getStreamStatus())) {
            System.out.println("流 " + streamName + " 当前状态为 " + streamDescription.getStreamStatus() + "，无法进行合并操作。");
            return;
        }

        List<Shard> shards = streamDescription.getShards();
        System.out.println("数据流中的分片数: " + shards.size());

        // 用于存储需要合并的分片
        List<Shard> shardsToMerge = new ArrayList<>();

        for (Shard shard : shards) {
            // 过滤掉已关闭的分片，只处理未关闭的分片
            if (shard.getSequenceNumberRange().getEndingSequenceNumber() == null) {
                shardsToMerge.add(shard);
            }
        }

        // 检查是否有需要合并的分片
        if (shardsToMerge.isEmpty()) {
            System.out.println("没有需要合并的分片。");
            return;
        }

        // 遍历需要合并的分片列表
        for (int i = 0; i < shardsToMerge.size() - 1; i += 2) {
            Shard shard1 = shardsToMerge.get(i);
            Shard shard2 = shardsToMerge.get(i + 1);

            // 合并相邻的两个分片
            MergeShardsRequest mergeShardsRequest = new MergeShardsRequest();
            mergeShardsRequest.setStreamName(streamName);
            mergeShardsRequest.setShardToMerge(shard1.getShardId());
            mergeShardsRequest.setAdjacentShardToMerge(shard2.getShardId());

            try {
                client.mergeShards(mergeShardsRequest);
                // 合并成功后，将第二个分片从列表中移除
                shardsToMerge.remove(i + 1);
            } catch (ResourceInUseException e) {
                System.out.println("合并分片时出现异常：" + e.getMessage());
            }
        }
    }



}
